package com.atguigu.gamll.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.atguigu.gamll.realtime.app.func.DimSink;
import com.atguigu.gamll.realtime.app.func.MyDebeziumDeserializationSchema;
import com.atguigu.gamll.realtime.app.func.TableProcessFunction;
import com.atguigu.gamll.realtime.beans.TableProcess;
import com.atguigu.gamll.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

// Flink 应用程序执行的基本流程
// - 基本环境准备
// - 检查点设置
// - source --> transform --> sink
public abstract class BaseApp {
    public void baseEntry() throws Exception {
        // TODO 1.基本环境准备
        // 1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 并行度
        env.setParallelism(4);

        // TODO 2.检查点设置
        // 2.1 开启检查点
//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
//        // 2.2 超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(6000L);
//        // 2.3 job取消，检查点是否保留
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        // 2.4 间隔时间
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        // 2.5 重启策略
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30), Time.milliseconds(3000L)));
//        // 2.6 状态后端
//        env.setStateBackend(new FsStateBackend("dfs://hadoop102:8020/gmall/ck"));
//        // 2.7 操作hdfs的用户
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        // TODO 3.不同的job有不同的业务逻辑
        biz(env);

        // TODO 4.提交job
        env.execute();
    }

    public abstract void biz(StreamExecutionEnvironment env);
}
